package com.tengju.data.infrastructure.kafka.consumer;

import com.alibaba.otter.canal.client.kafka.KafkaOffsetCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.tengju.data.domain.model.alert.AlertEvent;
import com.tengju.data.domain.model.alert.AlertService;
import com.tengju.data.domain.model.cdc.CDCEvent;
import com.tengju.data.domain.model.service.CDCEventHandleService;
import com.tengju.data.infrastructure.kafka.convert.EventConvert;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author chenhang
 * @Date 2021/9/29
 */
@Slf4j
public class HkUserBinlogConsumer {

    private ExecutorService executors;
    private String topics;
    private KafkaOffsetCanalConnector connector;

    private Properties properties;

    @Autowired
    private List<CDCEventHandleService> cdcEventHandleServiceList;

    @Autowired
    private AlertService alertService;

    @Value("${hkbinlog.resolve.flag}")
    private boolean binlogFlag;

    public void init() {
        connector = new KafkaOffsetCanalConnector(properties.getProperty("bootstrap.servers"), topics, null,properties.getProperty("group.id") , true);
        executors = Executors.newCachedThreadPool();
        executors.submit(() -> {
            try {
                connector.connect();
                connector.subscribe();
                while (true){
                    if (!binlogFlag) {
                        Thread.sleep(1000);
                        log.info("binlog解析暂停");
                        continue;
                    }
                    List<FlatMessage> messages = connector.getFlatList(100L, TimeUnit.MILLISECONDS);
                    if (messages == null){
                        continue;
                    }
                    boolean resolveSucceed = true;
                    for (FlatMessage record : messages) {
                        try {
                            dispatcher(record);
                        } catch (Exception e) {
                            AlertEvent event = new AlertEvent("binlog解析异常");
                            event.addItem("binlog信息",record.toString());
                            event.addItem("异常信息",e.getMessage());
                            alertService.alert(event);
                            connector.rollback();
                            Thread.sleep(100);
                            resolveSucceed = false;
                            break;
                        }
                    }
                    if (resolveSucceed) {
                        connector.ack();
                    }
                }
            } catch (Exception e) {
                log.error("数据处理失败", e);
                AlertEvent event = new AlertEvent("binlog解析异常,解析已终止");
                event.addItem("异常信息",e.getMessage());
                alertService.alert(event);
            }
        });
    }

    private void dispatcher(FlatMessage message) throws IOException {
        if (message == null) {
            return;
        }
        String eventType = message.getType();
        String tableName = message.getTable();
        if (eventType.equals(CanalEntry.EventType.QUERY.name()) || message.getIsDdl()) {
            log.info(" sql ----> " + message.getSql());
            return;
        }
        CDCEvent cdcEvent = EventConvert.binlogMsgToEvent(message);
        for (CDCEventHandleService cdcEventHandleService : cdcEventHandleServiceList) {
            if (!cdcEventHandleService.hitEvent(cdcEvent)) {
                continue;
            }
            if (eventType.equals(CanalEntry.EventType.DELETE.name())) {
                cdcEventHandleService.processDelete(cdcEvent);
            } else if (eventType.equals(CanalEntry.EventType.INSERT.name())) {
                cdcEventHandleService.processInsert(cdcEvent);
            } else if (eventType.equals(CanalEntry.EventType.UPDATE.name())) {
                cdcEventHandleService.processUpdate(cdcEvent);
            } else {
                log.warn("not support event type :  " + eventType + "message=" + cdcEvent);
            }
        }
    }

    public void destroy() {
        if (executors != null) {
            executors.shutdown();
        }
        try {
            if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
                log.error("Timeout.... Ignore for this case");
            }
        } catch (InterruptedException ignored) {
            log.error("Other thread interrupted this shutdown, ignore for this case.", ignored);
            Thread.currentThread().interrupt();
        }
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }

    public void setTopics(String topics) {
        this.topics = topics;
    }
}
